package com.elitel.frame.base.service.impl;

import com.elitel.common.kudu.KuduClientPoolUtil;
import com.elitel.frame.base.component.exportprojectdata.KuduConfInfo;
import com.elitel.frame.base.service.CfgDbService;
import com.elitel.frame.base.service.CfgTableService;
import com.elitel.frame.base.service.HiveService;
import com.elitel.frame.base.service.KuduService;
import com.elitel.frame.main.entity.CfgDb;
import com.elitel.frame.main.entity.CfgTable;
import com.elitel.frame.main.entity.vo.KeyValueModelVo;
import org.apache.commons.lang.StringUtils;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;

@Service
public class KuduServiceImpl implements KuduService {
    private static final Logger logger = LoggerFactory.getLogger(KuduServiceImpl.class);
//    @Value("#{'${kudu.databases}'.split(',')}")
//    private List<String> kuduDatabases;

    @Autowired(required = false)
    private KuduConfInfo kuduConfInfo;
    @Autowired(required = false)
    private HiveService hiveService;
    @Autowired
    private CfgTableService cfgTableService;
    @Autowired
    private CfgDbService cfgDbService;

    private KuduClientPoolUtil kuduPoolUtil = null;

    @PostConstruct
    public void initKuduPoolUtil() {
        if (kuduConfInfo == null) {
            return;
        }
        this.kuduPoolUtil = new KuduClientPoolUtil(kuduConfInfo.masterAddr,
                30000,
                10,
                60000,
                60000);
    }

    @Override
    public List<KeyValueModelVo> getAllFieldByTableName(String dbName, String tableName) {
        if (kuduPoolUtil == null) {
            return null;
        }
        KuduClient kuduClient = kuduPoolUtil.getClient();
        List<KeyValueModelVo> allFieldSingle = getAllFieldSingle(kuduClient, dbName, tableName);
        kuduPoolUtil.returnClient(kuduClient);
        return allFieldSingle;
    }

    @Override
    public List<String> getPrimaryKey(String dbName, String tableName) {
        if (kuduPoolUtil == null) {
            return null;
        }
        KuduClient kuduClient = kuduPoolUtil.getClient();
        List<String> pks = new ArrayList<>();
        try {
            KuduTable kuduTable = kuduClient.openTable(dbName + "." + tableName);
            Schema schema = kuduTable.getSchema();
            List<ColumnSchema> primaryKeyColumns = schema.getPrimaryKeyColumns();
            for (ColumnSchema pkColumn : primaryKeyColumns) {
                String pkColumnName = pkColumn.getName();
                pks.add(pkColumnName);
            }
        } catch (KuduException e) {
            e.printStackTrace();
        } finally {
            kuduPoolUtil.returnClient(kuduClient);
        }
        return pks;
    }

    @Override
    public List<String> getAllTables(String dbName) {
        if (kuduPoolUtil == null) {
            return null;
        }
        KuduClient kuduClient = kuduPoolUtil.getClient();
        List<String> tablesList = new ArrayList<>();
        try {
            List<String> list = kuduClient.getTablesList(dbName).getTablesList();
            for (String tableName : list) {
                if (tableName.startsWith(dbName + ".")) {
                    tableName = tableName.substring(tableName.indexOf(".") + 1);
                    tablesList.add(tableName);
                }
            }
        } catch (KuduException e) {
            e.printStackTrace();
        } finally {
            kuduPoolUtil.returnClient(kuduClient);
        }
        return tablesList;
    }

    @Override
    public void refreshKuduTableCount(String dbKey,String table_name) {
        List<Map<String, Object>> kuduAllTableCountList = getKuduTableCount(dbKey,table_name);

        int successCount = 0;
        List<String> failTables = new ArrayList<>();
        for (Map<String, Object> resultMap : kuduAllTableCountList) {
            String tableName = String.valueOf(resultMap.get("table_name"));
            long count = resultMap.get("table_count") != null ? (Long) resultMap.get("table_count") : 0L;

            int executeUpdate = cfgTableService.updateCfgtableCount(Integer.valueOf(String.valueOf(count)), tableName, dbKey);
            if (executeUpdate > 0) {
                successCount++;
            } else {
                failTables.add(tableName);
            }
        }
        logger.info("更新数据表总数：" + kuduAllTableCountList.size());
        logger.info("更新成功总数：" + successCount);
        logger.info("更新失败总数：" + (kuduAllTableCountList.size() - successCount));
        if (successCount != kuduAllTableCountList.size()) {
            logger.info("更新失败的表为：" + String.join(",", failTables));
        }
    }

//    /**
//     * 获取hdfs上存储的表数据量信息
//     *
//     * @return
//     */
//    private Long getHdfsTableCount(String dbName, String tableName) {
//        if (StringUtils.isBlank(dbName) || StringUtils.isBlank(tableName)) {
//            return 0L;
//        }
//        CfgDb cfgDb = new CfgDb();
//        cfgDb.setDbVersion("MYSQL");
//        cfgDb.setDbKey("metastore");
//        cfgDb.setDbDrivername(hiveMetastoreDbInfo.driverName);
//        cfgDb.setDbUrl(hiveMetastoreDbInfo.url);
//        cfgDb.setDbUser(hiveMetastoreDbInfo.userName);
//        cfgDb.setDbPwd(hiveMetastoreDbInfo.password);
//
//        String sqlFormat = "select b.name as dbname,a.TBL_NAME as tabname,c.PARAM_VALUE as numrows from TBLS a \n" +
//                "join DBS b on a.DB_ID=b.DB_ID and b.NAME='%s' and a.TBL_NAME='%s' \n" +
//                "join TABLE_PARAMS c on a.TBL_ID=c.TBL_ID and c.PARAM_KEY='numRows'";
//        List<Map<String, Object>> sqlResult = dataBaseMetadataService.getSqlResult(cfgDb, String.format(sqlFormat, dbName, tableName));
//        if (sqlResult != null) {
//            if (sqlResult.size() > 0) {
//                Map<String, Object> resMap = sqlResult.get(0);
//                if (resMap != null) {
//
//                    long count = resMap.get("numrows") != null ? Long.parseLong(resMap.get("numrows").toString()) : 0L;
//                    return count;
//                }
//            }
//        }
//        return 0L;
//    }

    @Override
    public void refreshSingleTableCount(CfgTable cfgTable) {
        if (cfgTable == null) {
            return;
        }
        String sqlFormat = "select count(*) as table_count from %s";
        Map<String, Object> resultMap = hiveService.queryForMap(String.format(sqlFormat, cfgTable.getDbName() + "." + cfgTable.getDtName()));
        Integer tableCount = resultMap.get("table_count") == null ? 0 : Integer.parseInt(String.valueOf(resultMap.get("table_count")));
//        Map<String, String> storageInfo = cfgTableService.getStorageInfo(cfgTable.getDbName() + '_' + cfgTable.getDtName());
//        if (storageInfo != null) {
//            String hdfsdb = storageInfo.get("hdfsdb");
//            String hdfstable = storageInfo.get("hdfstable");
//            Long hdfsTableCount = getHdfsTableCount(hdfsdb, hdfstable);
//            tableCount += Integer.parseInt(hdfsTableCount.toString());
//        }
        cfgTableService.updateCfgtableCount(tableCount, cfgTable.getDtName(), cfgTable.getDtConn());
    }

    @Override
    public Map<String, String> getPrimaryKeyMap(String dbName) {
        if (kuduPoolUtil == null) {
            return null;
        }
        List<String> allTables = getAllTables(dbName);
        KuduClient kuduClient = kuduPoolUtil.getClient();
        Map<String, String> pkListMap = new HashMap<>();
        for (String tableName : allTables) {
            try {
                KuduTable kuduTable = kuduClient.openTable(dbName + "." + tableName);
                Schema schema = kuduTable.getSchema();
                List<ColumnSchema> primaryKeyColumns = schema.getPrimaryKeyColumns();
                List<String> pks = new ArrayList<>();
                for (ColumnSchema pkColumn : primaryKeyColumns) {
                    String pkColumnName = pkColumn.getName();
                    pks.add(pkColumnName);
                }
                pkListMap.put(tableName, String.join(",", pks));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        kuduPoolUtil.returnClient(kuduClient);
        return pkListMap;
    }

    @Override
    public Map<String, List<KeyValueModelVo>> getAllFieldMap(String dbName) {
        if (kuduPoolUtil == null) {
            return null;
        }
        List<String> allTables = getAllTables(dbName);
        KuduClient kuduClient = kuduPoolUtil.getClient();
        Map<String, List<KeyValueModelVo>> resMap = new HashMap<>();
        for (String tableName : allTables) {
            try {
                List<KeyValueModelVo> keyValueModelVos = getAllFieldSingle(kuduClient, dbName, tableName);
                resMap.put(tableName, keyValueModelVos);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        kuduPoolUtil.returnClient(kuduClient);
        return resMap;
    }

    @Override
    public List<Map<String, Object>> getKuduTableCount(String dbKey, String table_mame) {
        CfgDb cfgDb = cfgDbService.selectByKey(dbKey);
        if (cfgDb == null) {
            return null;
        }
        String dbName = cfgDb.getDbName();
        List<String> tableNameList = new ArrayList<>();
        if (StringUtils.isBlank(table_mame)) {
            tableNameList = getAllTables(dbName);
        } else {
            tableNameList = Collections.singletonList(table_mame.contains(".") ? table_mame.substring(table_mame.indexOf(".") + 1) : table_mame);
        }

        List<Map<String, Object>> tableCountMap = new ArrayList<>();
        for (String tableName : tableNameList) {
            tableName = tableName.trim();
            String mdbName = dbName.trim();
            CfgTable cfgTable = cfgTableService.getInfoByDtconfigId(dbKey + "_" + tableName);

            if (cfgTable == null
//                        && !kuduDatabases.contains(dbName)
            ) {
                continue;
            }

            String sqlFormat = "select count(*) as table_count from %s";
            Map<String, Object> resultMap = hiveService.queryForMap(String.format(sqlFormat, mdbName + "." + tableName));
            resultMap.put("table_name", tableName);
            tableCountMap.add(resultMap);
        }
        return tableCountMap;
    }

    private List<KeyValueModelVo> getAllFieldSingle(KuduClient kuduClient, String dbName, String tableName) {
        List<KeyValueModelVo> keyValueModelVos = new ArrayList<>();
        try {
            KuduTable kuduTable = kuduClient.openTable(dbName + "." + tableName);
            Schema schema = kuduTable.getSchema();
            List<ColumnSchema> columns = schema.getColumns();
            for (ColumnSchema column : columns) {
                KeyValueModelVo keyValueModelVo = new KeyValueModelVo();
                String name = column.getName();
                String type = column.getType().getName();
                String comment = column.getComment();
                if (type.contains("int") || type.equals("unixtime_micros")) {
                    switch (type) {
                        case "int16":
                            type = "short";
                            break;
                        case "int64":
                            type = "long";
                            break;
                        case "unixtime_micros":
                            type = "timestamp";
                            break;
                        default:
                            type = "int";
                            break;
                    }
                }
                keyValueModelVo.setKey(name);
                keyValueModelVo.setValue(comment);
                keyValueModelVo.setDataType(type);
                keyValueModelVos.add(keyValueModelVo);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return keyValueModelVos;
    }
}
